package com.example.test;

import com.example.constant.KafkaTopic;
import com.example.entity.Order;
import com.example.entity.Source;
import com.example.entity.stream1.Source1;
import com.example.entity.stream10.Source10;
import com.example.entity.stream11.Source11;
import com.example.entity.stream12.Source12;
import com.example.entity.stream13.Source13;
import com.example.entity.stream14.Source14;
import com.example.entity.stream15.Source15;
import com.example.entity.stream16.Source16;
import com.example.entity.stream17.Source17;
import com.example.entity.stream18.Source18;
import com.example.entity.stream19.Source19;
import com.example.entity.stream2.Source2;
import com.example.entity.stream20.Source20;
import com.example.entity.stream3.Source3;
import com.example.entity.stream4.Source4;
import com.example.entity.stream5.Source5;
import com.example.entity.stream6.Source6;
import com.example.entity.stream7.Source7;
import com.example.entity.stream8.Source8;
import com.example.entity.stream9.Source9;
import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaSourceExample {


    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.80:9092,192.168.8.80:9093,192.168.8.80:9094");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
        props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://192.168.8.80:8081");

        Producer<String, Source1> producer1 = new KafkaProducer<>(props);
        Producer<String, Source2> producer2 = new KafkaProducer<>(props);
        Producer<String, Source3> producer3 = new KafkaProducer<>(props);
        Producer<String, Source4> producer4 = new KafkaProducer<>(props);
        Producer<String, Source5> producer5 = new KafkaProducer<>(props);
        Producer<String, Source6> producer6 = new KafkaProducer<>(props);
        Producer<String, Source7> producer7 = new KafkaProducer<>(props);
        Producer<String, Source8> producer8 = new KafkaProducer<>(props);
        Producer<String, Source9> producer9 = new KafkaProducer<>(props);
        Producer<String, Source10> producer10 = new KafkaProducer<>(props);
        Producer<String, Source11> producer11 = new KafkaProducer<>(props);
        Producer<String, Source12> producer12 = new KafkaProducer<>(props);
        Producer<String, Source13> producer13 = new KafkaProducer<>(props);
        Producer<String, Source14> producer14 = new KafkaProducer<>(props);
        Producer<String, Source15> producer15 = new KafkaProducer<>(props);
        Producer<String, Source16> producer16 = new KafkaProducer<>(props);
        Producer<String, Source17> producer17 = new KafkaProducer<>(props);
        Producer<String, Source18> producer18 = new KafkaProducer<>(props);
        Producer<String, Source19> producer19 = new KafkaProducer<>(props);
        Producer<String, Source20> producer20 = new KafkaProducer<>(props);
           int userId = 20;
            while (userId < 300){

                Source1 source1 = new Source1( userId);
                Source2 source2 = new Source2( userId);
                Source3 source3 = new Source3( userId);
                Source4 source4 = new Source4( userId);
                Source5 source5 = new Source5( userId);
                Source6 source6 = new Source6( userId);
                Source7 source7 = new Source7( userId);
                Source8 source8 = new Source8( userId);
                Source9 source9 = new Source9( userId);
                Source10 source10 = new Source10( userId);
                Source11 source11 = new Source11( userId);
                Source12 source12 = new Source12( userId);
                Source13 source13 = new Source13( userId);
                Source14 source14 = new Source14( userId);
                Source15 source15 = new Source15( userId);
                Source16 source16 = new Source16( userId);
                Source17 source17 = new Source17( userId);
                Source18 source18 = new Source18( userId);
                Source19 source19 = new Source19( userId);
                Source20 source20 = new Source20( userId);
                ProducerRecord<String, Source1> record1 = new ProducerRecord<>(KafkaTopic.sourceId1, String.valueOf(userId), source1);
                ProducerRecord<String, Source2> record2 = new ProducerRecord<>(KafkaTopic.sourceId2, String.valueOf(userId), source2);
                ProducerRecord<String, Source3> record3 = new ProducerRecord<>(KafkaTopic.sourceId3, String.valueOf(userId), source3);
                ProducerRecord<String, Source4> record4 = new ProducerRecord<>(KafkaTopic.sourceId4, String.valueOf(userId), source4);
                ProducerRecord<String, Source5> record5 = new ProducerRecord<>(KafkaTopic.sourceId5, String.valueOf(userId), source5);
                ProducerRecord<String, Source6> record6 = new ProducerRecord<>(KafkaTopic.sourceId6, String.valueOf(userId), source6);
                ProducerRecord<String, Source7> record7 = new ProducerRecord<>(KafkaTopic.sourceId7, String.valueOf(userId), source7);
                ProducerRecord<String, Source8> record8 = new ProducerRecord<>(KafkaTopic.sourceId8, String.valueOf(userId), source8);
                ProducerRecord<String, Source9> record9 = new ProducerRecord<>(KafkaTopic.sourceId9, String.valueOf(userId), source9);
                ProducerRecord<String, Source10> record10 = new ProducerRecord<>(KafkaTopic.sourceId10, String.valueOf(userId), source10);
                ProducerRecord<String, Source11> record11 = new ProducerRecord<>(KafkaTopic.sourceId11, String.valueOf(userId), source11);
                ProducerRecord<String, Source12> record12 = new ProducerRecord<>(KafkaTopic.sourceId12, String.valueOf(userId), source12);
                ProducerRecord<String, Source13> record13 = new ProducerRecord<>(KafkaTopic.sourceId13, String.valueOf(userId), source13);
                ProducerRecord<String, Source14> record14 = new ProducerRecord<>(KafkaTopic.sourceId14, String.valueOf(userId), source14);
                ProducerRecord<String, Source15> record15 = new ProducerRecord<>(KafkaTopic.sourceId15, String.valueOf(userId), source15);
                ProducerRecord<String, Source16> record16 = new ProducerRecord<>(KafkaTopic.sourceId16, String.valueOf(userId), source16);
                ProducerRecord<String, Source17> record17 = new ProducerRecord<>(KafkaTopic.sourceId17, String.valueOf(userId), source17);
                ProducerRecord<String, Source18> record18 = new ProducerRecord<>(KafkaTopic.sourceId18, String.valueOf(userId), source18);
                ProducerRecord<String, Source19> record19 = new ProducerRecord<>(KafkaTopic.sourceId19, String.valueOf(userId), source19);
                ProducerRecord<String, Source20> record20 = new ProducerRecord<>(KafkaTopic.sourceId20, String.valueOf(userId), source20);

                try {
                    producer1.send(record1);
                    producer2.send(record2);
                    producer3.send(record3);
                    producer4.send(record4);
                    producer5.send(record5);
                    producer6.send(record6);
                    producer7.send(record7);
                    producer8.send(record8);
                    producer9.send(record9);
                    producer10.send(record10);
                    producer11.send(record11);
                    producer12.send(record12);
                    producer13.send(record13);
                    producer14.send(record14);
                    producer15.send(record15);
                    producer16.send(record16);
                    producer17.send(record17);
                    producer18.send(record18);
                    producer19.send(record19);
                    producer20.send(record20);
                    producer1.flush();
                    producer2.flush();
                    producer3.flush();
                    producer4.flush();
                    producer5.flush();
                    producer6.flush();
                    producer7.flush();
                    producer8.flush();
                    producer9.flush();
                    producer10.flush();
                    System.out.println("Message sent successfully");
                    userId++;
                    Thread.sleep(1000);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {


                }
            }

            producer1.close();
            producer1.close();
            producer3.close();
            producer4.close();
            producer5.close();
            producer6.close();
            producer7.close();
            producer8.close();
            producer9.close();
            producer10.close();

    }

}